package com.spark

import java.text.SimpleDateFormat
import java.util.Calendar

import com.trs.hybase.hadoop.common.HybaseInputFormat
import org.apache.hadoop.conf.Configuration
import org.apache.spark.{SparkConf, SparkContext}

/**
  * Created by Administrator on 2017/7/28.
  */
object hybase {


  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("hybase").setMaster("local")
    val sc = new SparkContext(conf)
    val (startTime, endTime) = lastNDays(1)
    //val query = "DOCPUBTIME:[\"" + startTime + "\" TO \"" + endTime + "\"]"
    val query ="TIMESTAMP:[\"20170728173000\" TO *]"
    val tableName = "website_formal_20160321"
    val fields = "ZB_GUID_CHAR;CONTENT;DOCTITLE"
    val data = queryFromHybase(sc, tableName, fields, query)


    println(data.count())


  }


  def lastNDays(n: Int) = {
    val endTimeDateFormat = new SimpleDateFormat("yyyy.MM.dd HH:mm:ss")
    val beforeTimeDateFormat = new SimpleDateFormat("yyyy.MM.dd 00:00:00")
    val cal = Calendar.getInstance()
    val endTime = endTimeDateFormat.format(cal.getTime())
    // 默认的时间选择是当天
    var beforeTime = beforeTimeDateFormat.format(cal.getTime())
    cal.add(Calendar.DAY_OF_MONTH, 1 - n)
    beforeTime = beforeTimeDateFormat.format(cal.getTime())
    (beforeTime, endTime)
  }

  def queryFromHybase(ctx: SparkContext, dbTable: String, dbFields: String, query: String) = {
    val jobConf = new Configuration
    jobConf.set(HybaseInputFormat.HYBASE_DB_HOST, "hybase.zhaochao.net")
    jobConf.set(HybaseInputFormat.HYBASE_DB_PORT, "80")
    jobConf.set(HybaseInputFormat.HYBASE_DB_USERNAME, "admin")
    jobConf.set(HybaseInputFormat.HYBASE_DB_PASSWORD, "Trsadmin_2015_!@#")
    jobConf.set(HybaseInputFormat.HYBASE_DB_TABLE, dbTable)
    jobConf.set(HybaseInputFormat.HYBASE_DB_RECORDFIELDS, dbFields)
    jobConf.set(HybaseInputFormat.HYBASE_DB_QUERY, query)
    ctx.newAPIHadoopRDD(jobConf, classOf[com.trs.hybase.hadoop.common.HybaseInputFormat], classOf[org.apache.hadoop.io.LongWritable], classOf[com.trs.hybase.client.TRSRecord])
  }

}
